package com.jltx.zeus.observable

import io.reactivex.Observable
import io.reactivex.Observer
import io.reactivex.disposables.Disposable
import io.reactivex.exceptions.CompositeException
import io.reactivex.exceptions.Exceptions
import io.reactivex.plugins.RxJavaPlugins

/**
 * 将同步调用转换为 Observable
 *
 * @param <T>
</T> */
internal class SyncCallObservable<T>(private val mCall: SyncCall<T>) :
    Observable<T>() {
    override fun subscribeActual(observer: Observer<in T>) {
        val disposable = CallDisposable(mCall)
        observer.onSubscribe(disposable)
        if (disposable.isDisposed) {
            return
        }
        var terminated = false
        try {
            val result = mCall.execute()
            if (!disposable.isDisposed) {
                observer.onNext(result)
            }
            if (!disposable.isDisposed) {
                terminated = true
                observer.onComplete()
            }
        } catch (t: Throwable) {
            Exceptions.throwIfFatal(t)
            if (terminated) {
                RxJavaPlugins.onError(t)
            } else if (!disposable.isDisposed) {
                try {
                    observer.onError(t)
                } catch (inner: Throwable) {
                    Exceptions.throwIfFatal(inner)
                    RxJavaPlugins.onError(CompositeException(t, inner))
                }
            }
        }
    }

    private class CallDisposable internal constructor(private val mCall: SyncCall<*>) :
        Disposable {

        @Volatile
        private var mDisposed = false
        override fun dispose() {
            mDisposed = true
            mCall.cancel()
        }

        override fun isDisposed(): Boolean {
            return mDisposed
        }

    }

}